package cn.cihon.ubi.qorosauto.storm;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.FileInputStream;
import java.text.SimpleDateFormat;
import java.util.*;

import javax.jms.*;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.cihon.ubi.qorosauto.biz.VehicleStandardDataFactory;
import cn.cihon.ubi.qorosauto.bo.VehicleStandardData;

/*
 * 从ActiveMQ中取数据，取出的数据为二进制数据，每次取1024bit也就是一个字节
 * 取完数据后按byte发送到相应的Bolt
 */
public class ActiveMQSpout implements IRichSpout {

	private static final long serialVersionUID = 1L;
	private Logger log = LoggerFactory.getLogger(ActiveMQSpout.class);
	private SpoutOutputCollector outputCollector;
	public int i = 0; // 作为messageId
	boolean flag = false;
	public FileInputStream in = null;
	public DataInputStream dis = null;
	public MessageConsumer consumer = null;
	public int novinCnt = 0;

	public MessageConsumer getConsumer() throws Exception {

		Connection connection = null;
		// Session：一个发送或接收消息的线程
		Session session;
		// 消费者，消息接收者
		MessageConsumer consumer;

		String url = "tcp://123.57.54.113:61616";

		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD, url);

		// 构造从工厂得到连接对象
		connection = connectionFactory.createConnection();

		// connection.setClientID("id" + (int)(Math.random()*50) + 50);
		connection.setClientID("id99");
		// 启动
		connection.start();
		// 获取操作连接
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 获取session
		Topic topic = session.createTopic("carInfo_topic1");
		// 得到消息生成者【发送者】
		consumer = session.createDurableSubscriber(topic, "MQ_sub");
		return consumer;
	}

	public void nextTuple() {
		byte[] buffer = null;
		try {
			BytesMessage message = (BytesMessage) consumer.receiveNoWait();
			if (null != message) {

				buffer = new byte[1024];
				try {
					message.readBytes(buffer);
				} catch (Exception e) {
					e.printStackTrace();
				}
				
				this.outputCollector.emit(new Values(buffer), i++);     //发送二进制数据
				
			}

		} catch (Exception ex) {
			ex.printStackTrace();
		}

	}

	public void declareOutputFields(OutputFieldsDeclarer declare) {
		// TODO Auto-generated method stub
		declare.declare(new Fields("byte")); // 声明streamId
	}

	public void open(Map conf, TopologyContext context,
			SpoutOutputCollector collector) {
		try {
			consumer = getConsumer();
		} catch (Exception e) {
			e.printStackTrace();
		}
		this.outputCollector = collector;
	}

	public void activate() {
		// TODO Auto-generated method stub

	}

	public void deactivate() {
		// TODO Auto-generated method stub

	}

	public Map getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

	public boolean isDistributed() {
		// TODO Auto-generated method stub
		return false;
	}

	public void close() {
		// TODO Auto-generated method stub

	}

	public void ack(Object msgId) {
		log.info("正确ack!");
	}

	public void fail(Object msgId) {
		log.info("错误fail!");
	}

}
